package com.afdor.rws.event.channel.aliyun.mq.queue;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.channel.aliyun.mq.AbstractAliyunEventConsumer;
import com.afdor.rws.event.channel.client.consumer.EventConsumer;
import com.aliyun.mns.client.AsyncCallback;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.model.Message;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * 基于阿里云MQ发布订阅模式的事件消费者实现
 *
 * @author 悭梵
 * @date Created in 2018-08-15 16:25
 */
@Slf4j
@ToString
@EqualsAndHashCode
@Extension(value = "amq_queue_async", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventConsumer.class)
public class AliyunMQQueueAsyncEventConsumer extends AbstractAliyunEventConsumer implements AsyncCallback<List<Message>> {


    @Override
    protected boolean batchHandlerMessage() {
        CloudQueue cloudQueue = getCloudQueue();
        if (getWaitSeconds() > 0) {
            cloudQueue.asyncBatchPopMessage(getBatchSize(), getWaitSeconds(), this);
        } else {
            cloudQueue.asyncBatchPopMessage(getBatchSize(), this);
        }
        return Boolean.FALSE;
    }

    @Override
    protected boolean handlerMessage() {
        CloudQueue cloudQueue = getCloudQueue();
        AsyncCallback asyncCallback = new AsyncCallback<Message>() {
            @Override
            public void onSuccess(Message result) {
                getCloudQueue().deleteMessage(handlerMessage(result));
            }

            @Override
            public void onFail(Exception ex) {

            }
        };
        if (getWaitSeconds() > 0) {
            cloudQueue.asyncPopMessage(getWaitSeconds(), asyncCallback);
        } else {
            cloudQueue.asyncPopMessage(asyncCallback);
        }
        return Boolean.FALSE;
    }

    @Override
    public void onSuccess(List<Message> messageList) {
        if (messageList == null || messageList.isEmpty()) {
            return;
        } else {
            List<String> receiptHandleList = new ArrayList<>(messageList.size());
            messageList.parallelStream().forEach(message -> {
                String receiptHandle = handlerMessage(message);
                if (StringUtils.isNotBlank(receiptHandle)) {
                    receiptHandleList.add(receiptHandle);
                }
            });
            getCloudQueue().batchDeleteMessage(receiptHandleList);
        }
    }

    @Override
    public void onFail(Exception ex) {
        log.error("异步从阿里云MNS获取消息失败", ex);
    }
}
